package com.example.flink.wordcount;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordDataSetJob {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        DataStream<String> text = env.fromElements(
                "flink aprk storm",
                "flink saprk storm",
                "flink aprk storm"
        );

        DataStream<Tuple2<String,Integer>> wordcounts =
                text.flatMap(new wordCount()).name("flatmap")
                        .keyBy(0).sum(1).name("counter-token");

        wordcounts.print().name("print-sink");
        env.execute("WordCount");

    }
}
